package com.qf.rector;

import com.sun.org.apache.bcel.internal.generic.NEW;
import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.CoreSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.ParallelFlux;

import java.io.Serializable;
import java.sql.Array;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CountDownLatch;

/*
 *@Description:
 *@author:刘哥
 *@Date:86180 2022/9/21 16:17
 * 响应式编程:
 *            响应流必须是无阻塞的。
 *            响应流必须是一个数据流。
 *            它必须可以异步执行。
 *            并且它也应该能够处理背压
 */
public class FluxTest {
    /*
    以编程方式创建具有多次发射能力的Flux,元素通过FluxSink API以同步或异步方式进行。
     */
    @Test
    public void fluxCreat() {
//        创建一个生产者
        Flux<Object> flux = Flux.create(fluxSink -> {
            fluxSink.next("第一个数据")
                    .next("第二个数据")
                    .next("我鼠标垫被偷了")
                    .complete();//complete完成了
        });
//        flux.subscribe();//忽略数据订阅,只是单纯的运行方法
//        System.out::println=System.out.println(flux);
//        flux.subscribe(System.out::println);//通过consumer将吗，每一个数据传入进行输出
//       里面放入一个订阅者对象
        flux.subscribe(new Subscriber<Object>() {
            //            Subscription subscription承接上下文对象,和application对象相似
            @Override
            public void onSubscribe(Subscription subscription) {
                subscription.request(3);//获取生产者的那几个数据
            }

            //            当subscription获取数据时，就会执行onNext,
//            o 时上面传过来的数据
            @Override
            public void onNext(Object o) {
                System.out.println("o = " + o);
//                int i=10/0;
            }

            /*
                当发生错误时才会执行这个方法
             */
            @Override
            public void onError(Throwable throwable) {
                System.out.println("又错了,你错哪里了");
            }

            //            最后执行 的方法,即订阅完成后
            @Override
            public void onComplete() {
                System.out.println("onComplete方法执行了  ");
            }
        });
    }

    /*
    以编程方式创建一个的Flux,通过consumer回调逐一生成信号；generate中next只能调1次，否则会报错
     */
    @Test
    public void testGenrate() {
        Flux<Object> flux = Flux.generate(synchronousSink -> {
//            next方法只能在此调用一次，必须调用complete()方法表示完成,不然会变成无限流,引用不会停止
            synchronousSink.next("申志恒");
//            synchronousSink.next("师冠雄");
            synchronousSink.complete();
        });
//        订阅
        flux.subscribe(System.out::print);
    }

    /*
    通过我们指定的数据来创建一个flux
     */
    @Test
    public void testJust() {
        Flux<String> just = Flux.just("申志恒", "赵健", "师冠雄", "刘老四");
        just.subscribe(System.out::print);
    }

    /*
    用Flux API指定的Publisher创建一个Flux
     */
    @Test
    public void tesFrom() {
//        Flux.from(里面调用生产者对象)
        Flux<? extends Serializable> flux = Flux.from(Flux.just("1", 2, 3, 4, 5, 6, 7, 8, 9));
        flux.subscribe(new Subscriber<Object>() {
            //            创建订阅上下文对象
            private Subscription subscription;

            @Override
            public void onSubscribe(Subscription subscription) {
                subscription.request(2);
            }

            @Override
            public void onNext(Object o) {
                System.out.println("o = " + o);
//                subscription.request(3);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onComplete() {
                System.out.println("结束");
            }
        });

    }

    @Test
    public void testFromArray() {
//        通过数组创建
        Flux<String> stringFlux = Flux.fromArray(new String[]{"申志恒", "赵健", "师冠雄", "刘老四"});
        stringFlux.subscribe(System.out::print);
    }

    @Test
    public void testFromIterable() {
        HashSet<Object> hashSet = new HashSet<>();
        hashSet.add("李易峰");
        hashSet.add("吴亦凡");
        hashSet.add("柯震东");
        hashSet.add("房祖名");
        hashSet.add("张桓玮偷我的鼠标垫");
        Flux<Object> flux = Flux.fromIterable(hashSet);
        flux.subscribe(System.out::print);
    }

    @Test
    public void testFromStream() {
        HashSet<Object> hashSet = new HashSet<>();
        hashSet.add("花自飘零水自流,一种相思,两处闲愁,此情无计可消除,才下眉头,却上心头");
        hashSet.add("昨夜西风凋碧树,独上高楼,望尽天涯路");
        hashSet.add("雨打梨花深闭门,忘了青春,误了青春");
        hashSet.add("醉后不知天在水,满船清梦压星河");
        hashSet.add("让我将心儿摘下,试着油锅慢慢煎炸");
//      需要通过一个stream流
        Flux<Object> flux = Flux.fromStream(hashSet.stream());
        flux.subscribe(System.out::print);
    }

    @Test
    public void defer() {
        Flux.defer(() -> Flux.just("just", "just1", "just2"))
                .subscribe(System.out::print);
    }

    //    循环生成数据,从x开始自增,间隔时间为指定的时间
    @Test
    public void testInterVal() throws InterruptedException {
//        Duration有效期
        Flux.interval(Duration.of(2, ChronoUnit.SECONDS))
                .subscribe(new Subscriber<Long>() {
                    private Subscription subscription;
                    private long startTime;

                    @Override
                    public void onSubscribe(Subscription subscription) {
                        this.subscription = subscription;
                        subscription.request(1);
                        startTime = System.currentTimeMillis();
                    }

                    @Override
                    public void onNext(Long aLong) {
                        long currentTimeMillis = System.currentTimeMillis();
                        System.err.println(aLong);
                        System.err.println(currentTimeMillis - startTime);
                        startTime = currentTimeMillis;
                        subscription.request(1);
                    }

                    @Override
                    public void onError(Throwable throwable) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
        //防止程序过早退出，放一个CountDownLatch拦住
        CountDownLatch latch = new CountDownLatch(1);
        latch.await();

    }

    @Test
    public void testRange() {
//       range(10，20)第一个参数开始值,第二个是结束值,默认使用背压,每一次调用都是重新调用
        Flux.range(10, 20).subscribe(new Subscriber<Integer>() {
            private Subscription subscription;

            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1);
            }

            @Override
            public void onNext(Integer integer) {
                System.err.println("FluxTest中的方法onNext 执行了--->");
                try {
                    Thread.sleep(1000);
                    System.err.println(integer);
                    subscription.request(1);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    /*
     Reactor 发射器转换操作函数
     */
    @Test
    public void testConcatWithValue() {
//        将值连接到Flux的末尾,需要在第一个flux发出complete之后才追加
        Flux.just("a", "b", "c").concatWithValues("d", "e", "f").subscribe(System.out::print);
    }

    @Test
    public void testConcat() {
        Flux.concat(Flux.just("a", "b", "c").doOnNext(s -> {
                            System.err.println("马上发出去的数据是" + s);
                            if (new Random().nextBoolean()) {
                                int i = 1 / 0;
                            }
                        })//在订阅者的onNext方法执行之前先执行
                        , Flux.just("1", "2", "3"))
                .subscribe(new Subscriber<String>() {
                    private Subscription subscription;

                    @Override
                    public void onSubscribe(Subscription subscription) {
                        this.subscription = subscription;
                        subscription.request(1);
                    }

                    @Override
                    public void onNext(String s) {
                        System.out.println("onNext方法执行了");
                        subscription.request(1);
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        System.out.println("发生异常了");
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("完成了");
                    }
                });

    }

    /*
    此操作符在取消时丢弃内部排队的元素以产生背压。
     */
    @Test
    public void testconcatDelayError() {
        Flux.concatDelayError(Flux.just("a", "b", "c").doOnNext(s -> {
                            System.err.println("马上发出去的数据是" + s);
                            if (new Random().nextBoolean()) {
                                int i = 1 / 0;
                            }
                        })//在订阅者的onNext方法执行之前先执行
                        , Flux.just("1", "2", "3"))
                .subscribe(new Subscriber<String>() {
                    private Subscription subscription;

                    @Override
                    public void onSubscribe(Subscription subscription) {
                        this.subscription = subscription;
                        subscription.request(1);
                    }

                    @Override
                    public void onNext(String s) {
                        System.out.println("onNext方法执行了");
                        subscription.request(1);
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        System.out.println("发生异常了");
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("完成了");
                    }
                });
    }

    /*
    交替获取两个flux中的数据，返回一个合并的Flux
     */
    @Test
    public void testMeger() {
        Flux.merge(Flux.interval(Duration.ofMillis(0), Duration.ofMillis(100)).take(5), Flux.interval(Duration.ofMillis(50), Duration.ofMillis(100)).take(3))
                .toStream()
                .forEach(System.out::println);
    }

    /*
    返回 一个按照顺序排序合并Flux，可以自己设置排序方法
     */
    @Test
    public void testmergeComparing() {
        // Flux.mergeComparing(Flux.just(1,3,5),Flux.just(2,4,6)).subscribe(System.out::println);
//        Flux.mergeComparing((obj1,obj2)->1,Flux.just(1,3,5),Flux.just(2,4,6)).subscribe(System.out::println);
        Flux.mergeComparing((obj1, obj2) -> obj2 - obj1, Flux.just(1), Flux.just(3), Flux.just(5), Flux.just(2), Flux.just(4), Flux.just(6)).subscribe(System.out::println);

    }

    @Test
    public void testZip() {
        //默认就是将两个flux中的每个数据按照顺序合并到一个数组中
        //   Flux.zip(Flux.just(1,3,5),Flux.just(2,4,6)).subscribe(System.out::println);
//        按照自己指定个规则,合并到一个数组中
        Flux.zip(Flux.just(1, 3, 5), Flux.just(2, 4, 6), (int1, int2) -> int1 - int2).subscribe(System.out::println);

    }

    //   判断生产者的参数是否都满足条件
    @Test
    public void testAlll() {
        Flux.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).all(integer -> integer > 0).subscribe(System.out::print);
        Flux.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).all(integer -> integer > 5).subscribe(System.out::print);
    }

    /*
     判断生产者中任意一个参数满足条件即可
     */
    @Test
    public void testAny() {
        Flux.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).any(integer -> integer > 0).subscribe(System.out::print);
        Flux.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).any(integer -> integer > 5).subscribe(System.out::print);
    }

    /*
    将此Flux转换为目标类型。将所有的数据挨个传递给我们,我们最终将数组进行响应的操作后返回一个结果,具体返回什么取决于我们自己的业务
     */
    @Test
    public void testAs() {
        Integer as = Flux.just(1, 2, 3, 4).as(integerFlux -> {
            Integer[] s = {0};
            integerFlux.subscribe(integer -> s[0] = s[0] + integer);
            return s[0];
        });
        System.out.println("as = " + as);
    }

    /*
    将所有传入的值收集到一个列表缓冲器中，一旦该Flux完成，将由返回的Flux发出。
     */
    @Test
    public void testBuffer() {
//        设置缓存器中一次最大放入几个参数,多出的在次按照设置分组,知道全部放入为止
        Flux.just(1, 2, 3, 4, 5).buffer(4).subscribe(System.out::print);
//      buffer(3，1)第二个skip代表每次分组跳过上一组几个开始为下一组的起始值,直到分完为止
        Flux.just(1, 2, 3, 4, 5).buffer(3, 1).subscribe(System.out::println);
    }

    /*
    将flux数据按照设置的个数放入缓存里面,设置缓存的有效期,下次读取自动从缓存里面读取
     */
    @Test
    public void testCache() throws InterruptedException {
        Flux<String> flux = Flux.just("a", "b", "c", "d", "e").cache(3, Duration.of(10, ChronoUnit.SECONDS));
        flux.subscribe(System.out::print);
        Thread.sleep(5000);
        flux.subscribe(System.out::print);
        Thread.sleep(5000);
        flux.subscribe(System.out::print);
        Thread.sleep(5000);
        flux.subscribe(System.out::print);
    }

    /*
    将产生的Flux类型强制转换为目标产生类型的Flux,注意类型转换异常,一般是因为数据的来源中没有具体的类型,所以是object,但是我们在操作的时候是知道具体类型的,要么每次强转数据,要么我们可以通过cast一次性强转再进行操作
     */
    @Test
    public void testCast() {
        Flux<String> flux = Flux.just(1, 2, 3, 4, 5).cast(String.class);
        flux.subscribe(System.out::println);
    }

    @Test
    public void testCollect() {
        Flux.just(1, 2, 3, 4, 5, 6, 6).collect(
                () -> "a",//将这个返回的a拼接到我们的每个数据的前面,这个只会执行一次,并且当前的就是最终的返回结果
                (t, u) -> System.out.println(t + u)//t就是我们每次拼接到前面的那个a, u就是每次遍历的数据
        ).subscribe(System.err::println);//订阅其实就是我们返回的那个a
    }

    /*
    如果没有任何数据完成此序列，则提供默认的唯一值
     */
    @Test
    public void testDefaultIfEmpty() {
//       collectList 将数据全部放到一个集合中之后作为一个返回值,放到mono中
        //        Mono<List<Integer>> mono = Flux.just(1, 2, 3, 4, 5).collect(Collectors.toList());
//        mono.subscribe(System.out::println);
//            defaultIfEmpty里面的参数时当flux是null时,为参数设置默认值,不位null就不设置
        Flux.empty().defaultIfEmpty("morenzhi").subscribe(System.out::println);
        Flux.just("1").defaultIfEmpty("morenzhi").subscribe(System.out::println);
    }

    /*
 对于每一个Subscriber，跟踪已经从这个Flux 跟踪元素和过滤出重复。 值本身被记录到一个用于检测的哈希集中。如果希望使用distinct(Object::hashcode)更轻量级的方法，该方法不保留所有对象，但是更容易由于hashcode冲突而错误地认为两个元素是不同的。
     */
    @Test
    public void testDistinct() {
//        Flux.just(5,1,2,5,6,12,4,1,5,2,8,8,4,5).distinct().subscribe(System.out::print);
//       当两个或多个相连的并且重复的数据改变时去重:如122454445去重后是124545
//        Flux.just(5,1,2,5,6,12,4,1,5,2,8,8,4,5).distinctUntilChanged().subscribe(System.out::print);
//        只保留第一个满足条件和第一个不满足条件的其他的满足条件和不满足条件的都认为重复的
        Flux.just(5, 1, 2, 5, 6, 12, 4, 1, 5, 2, 8, 8, 4, 5).distinct(integer -> integer > 5).subscribe(System.out::print);
    }

    /*
           当flux满足某个状态时进行相应操作
     */
    @Test
    public void testDonOn() {
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 1, 1, 2, 3)
                .doOnNext(integer -> System.out.println(integer))//当订阅者的onNext方法执行之前执行
                .doOnComplete(() -> System.out.println("doOnComplete"))//当Flux完成时触发附加行为（side-effect）。
                .doOnCancel(() -> System.out.println("doOnCancel"))//当Flux被取消时触发附加行为（side-effect）。
                .doOnError(throwable -> throwable.printStackTrace());//当发生错误时触发附加行为
        ParallelFlux<Integer> parallel = flux.parallel();

        parallel.subscribe(new Subscriber<Integer>() {
            private Subscription subscription;

            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1);
            }

            @Override
            public void onNext(Integer integer) {
                System.err.println("收到的数据" + integer + "--->" + Thread.currentThread().getName());

                int i = 10 / integer;
                subscription.request(1);

            }

            @Override
            public void onError(Throwable throwable) {
                System.err.println("发生异常了");
            }

            @Override
            public void onComplete() {
                System.err.println("jieshu了");
            }
        });
    }
    /*
    只保留符合断言条件的数据
     */
    @Test
    public void testFilter(){
        Flux.just(1, 2, 3, 4, 5, 1, 1, 2, 3).filter(integer -> integer>=4).subscribe(System.out::print);
    }
    /*
    让这个Flux完成，然后播放信号提供的Mono。 换句话说，忽略这个Flux和转换完成信号的发射和提供Mono完成信号。在得到的Mono中重放错误信号。
     */
    @Test
    public  void  testThen(){
        Flux.just(1, 2, 3, 4, 8, 5, 6, 6, 10, 100).doOnNext(integer -> {
            System.out.println("当前在到了参数"+integer+"的位置");
        }).doOnComplete(() -> System.out.println("完成了"))
//                在lflux结束后重新发送信号
                .then(Mono.just("abc")).subscribe(System.out::print);
    }
    @Test
    public void testFlagMap(){
        List<User> users = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            User user = new User();
            user.setId(i);
            user.setUserName("username-->"+i);
            ArrayList<Item> items = new ArrayList<>();
            for (int i1 = 0; i1 < 10; i1++) {
                Item item = new Item();
                item.setId(i1);
                item.setName(i+"-->"+i1+"==>itemname");
                items.add(item);
            }
            user.setItems(items);
            users.add(user);
        }
//        flagMap:扁平化转换,更多的使用场景是我们有一个集合(多个序列)数据,然后里面的每个数据内部又有一个集合数据,我们想将所有的数据的内部的集合的数据解开将里面的数据放到单独的flux对象中,
        Flux.fromIterable(users)
                .flatMap(user -> Flux.fromIterable(user.getItems()))
                .subscribe(System.out::print);
//       将Flux中的数据转换为另外一种格式的数据,并不会将集合里面的集合解开,只是将集合中的集合转换了一下类型。
        Flux.fromIterable(users)
                .map(user -> Flux.fromIterable(user.getItems()))
                .subscribe(System.out::print);
    }

}
